package com.chukun.flink.dataset.helloworld;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author chukun
 * @version 1.0.0
 * @description fink批处理helloworld
 * @createTime 2022年05月28日 19:16:00
 */
public class WordCount {

    public static final String [] WORDS = new String[]{
            "com.chukun.flink.dataset.helloworld.flink.start",
            "com.chukun.flink.dataset.helloworld.flink.fun",
            "com.chukun.flink.dataset.helloworld.flink.word.count",
            "com.chukun.flink.dataset.helloworld.flink.dataset",
    };

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String> source = env.fromElements(WORDS);

        DataSet<Tuple2<String, Integer>> mappedSource = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String word, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] tokens = word.toLowerCase().split("\\.");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        collector.collect(Tuple2.of(token, 1));
                    }
                }
            }
        });

        DataSet<Tuple2<String, Integer>> sumSet = mappedSource.groupBy("f0").sum(1);

        sumSet.print("hello-dataset");

        env.execute("WordCount");

    }
}
